package iwb.utils.kafkapatition;

import com.google.common.collect.Lists;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.CountDownLatch;


@Slf4j
public abstract class ListPartitionExec extends PartitionExecAbstract {

    /**
     * consumer.
     *
     * @param messager messager
     */
    @Override
    public void consumer(Messager messager) throws InterruptedException {
        List<List<Integer>> dataLists = Lists.partition(messager.getData(), 100);
        CountDownLatch latch = new CountDownLatch(dataLists.size());
        dataLists.forEach(list ->
                getExecutorService().submit(() -> {
                    runList(list);
                    latch.countDown();
                })
        );
        latch.await();
    }


    /**
     * 处理单个list的方法，需要继承类去实现。
     *
     * @param list
     */
    public abstract void runList(List list);

    public boolean isCompress(){
        return false;
    }
}
